package org.zhadoop.stream.cql.executor;

import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zhadoop.stream.api.Application;
import org.zhadoop.stream.cql.exception.ExecutorException;
import org.zhadoop.streaming.config.StreamingConfig;

import com.google.common.collect.Maps;

/**
 * 执行计划生成器
 * 
 * @author zwd
 *
 */
public class ExecutorPlanGenerator {
	private static final Logger LOG = LoggerFactory.getLogger(ExecutorPlanGenerator.class);

	private org.zhadoop.streaming.application.Application executorApp;

	private Map<String, String> systemConfig = Maps.newHashMap();

	private Application apiApplication = null;

	/**
	 * 生成执行计划 主要内容包含： 用户自定义的处理 执行计划组装 表达式解析 构建application
	 * <p/>
	 * 主要步骤： 1、解析所有的Schemas，构建schema信息 2、解析所有的Operator，构建OperatorInfo
	 * 3、整理Operator中的上下级关系 4、检查Application 5、提交Application
	 * 
	 * @param vap
	 * @return
	 */
	public org.zhadoop.streaming.application.Application generate(Application vap) throws ExecutorException {
		LOG.info("start to generator executor application for app" + vap.getApplicationId());
		apiApplication = vap;
		createEmptyApplication(vap.getApplicationId());
		parseUserDefineds(vap);
		parseSchemas();
		parseOperators();
		return executorApp;

	}

	private void createEmptyApplication(String applicationId) throws ExecutorException {
		Map<String, String> appConf = apiApplication.getConfs();
		StreamingConfig applicationConfig = getApplicationConfig(appConf);
		try {
			executorApp = ApplicationFactory.createApplication(applicationConfig, appid);
		} catch (StreamingException e) {
			ExecutorException.wrapStreamingException(e);
		}

	}

	private StreamingConfig getApplicationConfig(Map<String, String> applicationConfig) {
		StreamingConfig conf = new StreamingConfig();
		if(applicationConfig != null){
			conf.putAll(applicationConfig);
		}
		return conf;
	}

}
